package com.yqd.lesson.actor.nine

import akka.actor.{Actor, ActorPath, ActorSystem, Props}
import akka.persistence.{AtLeastOnceDelivery, PersistentActor}
import com.yqd.lesson.actor.four.Sender

case class Msg(deliveryId: Long, s: String)
case class Confirm(deliveryId: Long)

sealed trait Evt2
case class MsgSent(s: String) extends Evt2
case class MsgConfirmed(deliveryId: Long) extends Evt2

class AtLeastOnceActor(destination: ActorPath) extends PersistentActor with AtLeastOnceDelivery {

  override def persistenceId: String = "persistence-id"

  def receiveCommand: Receive = {
    case s: String           => persist(MsgSent(s))(updateState)
    case Confirm(deliveryId) => persist(MsgConfirmed(deliveryId))(updateState)
  }

  def receiveRecover: Receive = {
    case evt: Evt2 => {
      updateState(evt)
    }
  }

  def updateState(evt: Evt2): Unit = evt match {
    case MsgSent(s) => {
      println(s"msgsent $s code")
      deliver(destination)(deliveryId => Msg(deliveryId, s))
    }
    case MsgConfirmed(deliveryId) => confirmDelivery(deliveryId)
  }
}

class MyDestination extends Actor {
  def receive = {
    case Msg(deliveryId, s) =>
      //TODO业务处理
      println(s"$s is sended by $deliveryId")
      if(s.equals("exception")){
        throw new Exception;
      }
      sender() ! Confirm(deliveryId)
  }
}

object AtLeastOnceActor extends App{
  val system = ActorSystem("akka")
  val destination = system.actorOf(Props[MyDestination],"destination")
  val persistentActor = system.actorOf(Props(new AtLeastOnceActor(destination.path)))

  persistentActor ! "a"
  persistentActor ! "b"
  persistentActor ! "c"
  Thread.sleep(1000)
  //exception 会一直推送，因为目的Actor不会接收这个。
  persistentActor ! "exception"
  Thread.sleep(1000)
  persistentActor ! "print"

  Thread.sleep(10000)
  //system.terminate()
}
